"""
Case Type   : 发布订阅
Case Name   : 发布订阅过程中发布端网卡闪断
Create At   : 2022-11-10
Owner       : peilinqian
Description :
    1.两个集群创建表
    2.创建发布订阅
    3.发布端注入网卡闪断故障
    4.网卡闪断过程中发布端更新数据
    5.故障完成后订阅端查询数据是否更新
    6.订阅端注入网卡闪断故障
    7.网卡闪断过程中发布端更新数据
    8.故障完成后订阅端查询数据是否更新
    9.清理环境
Expect      :
    1.成功
    2.成功
    3.成功
    4.成功
    5.更新
    6.成功
    7.成功
    8.更新
    9.成功
History     : 
"""

import os
import time
import unittest

from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.Logger import Logger
from yat.test import Node
from yat.test import macro


class Pubsubclass(unittest.TestCase):
    def setUp(self):
        self.log = Logger()
        self.log.info("----this is setup----")
        self.log.info(f"----{os.path.basename(__file__)} start----")
        self.pri_pub_dbuser = Node(node='PrimaryDbUser')
        self.pri_pub_root = Node(node='PrimaryRoot')
        self.pri_sub_dbuser = Node(node='remote1_PrimaryDbUser')
        self.pri_sub_root = Node(node='remote1_PrimaryRoot')
        self.pub_sh = CommonSH('PrimaryDbUser')
        self.sub_sh = CommonSH('remote1_PrimaryDbUser')
        self.com = Common()
        self.constant = Constant()
        self.tb_name = 't_pub_sub_relia_001'
        self.sub_name = "s_pub_sub_relia_0001"
        self.pub_name = "p_pub_sub_relia_0001"
        self.parent_path_pub = os.path.dirname(macro.DB_INSTANCE_PATH)
        self.parent_path_sub = os.path.dirname(macro.DB_INSTANCE_PATH_REMOTE1)
        self.port = str(int(self.pri_pub_dbuser.db_port) + 1)
        self.user_param_pub = f'-U {self.pri_pub_dbuser.db_user} ' \
            f'-W {self.pri_pub_dbuser.db_password}'
        self.user_param_sub = f'-U {self.pri_sub_dbuser.db_user} ' \
            f'-W {self.pri_sub_dbuser.db_password}'

        self.log.info("----发布端wal_level初始值----")
        self.init_wal_level = self.pub_sh.show_param("wal_level;")

        self.log.info("----备份发布端pg_hba配置文件----")
        cmd = f"cp {os.path.join(macro.DB_INSTANCE_PATH, 'pg_hba.conf')} " \
            f"{os.path.join(self.parent_path_pub, 'pg_hba.conf')};"
        self.log.info(cmd)
        result = self.pri_pub_dbuser.sh(cmd).result()
        self.log.info(result)

        text = "--预置条件,修改发布端pg_hba expect:成功--"
        self.log.info(text)
        guc_res = self.pub_sh.executeGsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, '',
            f'host    replication  {self.pri_sub_dbuser.db_user} '
            f'{self.pri_sub_dbuser.db_host}/32 sha256')
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败' + text)
        guc_res = self.pub_sh.executeGsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '',
            'all', False, False, '',
            f'host    all  {self.pri_sub_dbuser.db_user} '
            f'{self.pri_sub_dbuser.db_host}/32 sha256')
        self.log.info(guc_res)
        self.assertTrue(guc_res, '执行失败' + text)

        text = "--预置条件,修改发布端wal_level expect:成功--"
        result = self.pub_sh.executeGsguc('set',
                                          self.constant.GSGUC_SUCCESS_MSG,
                                          'wal_level=logical')
        self.assertTrue(result, '执行失败' + text)
        text = "--重启发布端集群 expect:成功--"
        stop_msg = self.pub_sh.stopDbCluster()
        start_msg = self.pub_sh.startDbCluster()
        self.assertTrue(stop_msg, '执行失败' + text)
        self.assertTrue(start_msg, '执行失败' + text)

        self.log.info('----查询发布端主备状态----')
        self.pub_sh.getDbClusterStatus('status')
        self.log.info('----查询订阅端主备状态----')
        self.sub_sh.getDbClusterStatus('status',
                                       env_path=macro.DB_ENV_PATH_REMOTE1)

    def test_pub_sub(self):
        text = "----step1:在两个集群创建表 expect:成功----"
        self.log.info(text)
        self.log.info("--发布端建表--")
        sql_cmd = f"drop table if exists {self.tb_name};" \
            f"create table {self.tb_name}(c_int int primary key,c_text text);"
        pub_msg = self.pub_sh.executDbSql(sql_cmd,
                                          cmd=self.user_param_pub)
        self.assertEqual(pub_msg.count(self.constant.TABLE_CREATE_SUCCESS),
                         2, '执行失败' + text)
        self.log.info("--订阅端建表--")
        sub_msg = self.sub_sh.executDbSql(sql_cmd,
                                          dbname=None,
                                          cmd=self.user_param_sub,
                                          env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertEqual(sub_msg.count(self.constant.TABLE_CREATE_SUCCESS),
                         2, '执行失败' + text)

        text = "----step2:创建发布订阅 expect:成功----"
        self.log.info(text)
        self.log.info("--创建发布--")
        sql_cmd = f"drop publication if exists {self.pub_name};" \
            f"create publication {self.pub_name} for all tables;"
        pub_msg = self.pub_sh.executDbSql(sql_cmd,
                                          cmd=self.user_param_pub)
        self.assertIn(self.constant.create_pub_succ_msg, pub_msg,
                      '执行失败' + text)

        self.log.info("--订阅端生成加密文件--")
        result = self.sub_sh.execute_generate(macro.PASSWD_INITIAL,
                                              env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertIn(self.constant.create_keycipher_success, result,
                      '执行失败' + text)
        self.log.info("--创建订阅--")
        sql_cmd = f"drop subscription if exists {self.sub_name};" \
            f"create subscription {self.sub_name} connection " \
            f"'host={self.pri_pub_dbuser.db_host} " \
            f"port={self.port} user={self.pri_pub_dbuser.db_user} " \
            f"dbname={self.pri_pub_dbuser.db_name} " \
            f"password={self.pri_pub_dbuser.ssh_password}' publication " \
            f"{self.pub_name};"
        sub_msg = self.sub_sh.executDbSql(sql_cmd, dbname=None,
                                          cmd=self.user_param_sub,
                                          env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertIn(self.constant.create_sub_succ_msg, sub_msg, '执行失败'
                      + text)

        text = "----step3:发布端注入网卡闪断故障 expect:成功----"
        self.log.info('----获取网卡----')
        network_cmd = f'ifconfig |grep {self.pri_pub_root.db_host} -B 1'
        self.log.info(network_cmd)
        network_msg = self.pri_pub_root.sh(network_cmd).result()
        network_card = network_msg.splitlines()[0].split(':')[0].strip()
        self.log.info(network_card)
        self.log.info("----故障注入----")
        inject_cmd = f"rNetLink_flash (repeat,duration,interval,dev) " \
            f"values(3,5,5,{network_card})"
        inject_result = self.com.cfe_inject(self.pri_pub_root, inject_cmd)
        self.log.info(inject_result)
        self.assertIn(self.constant.cfe_inject_netflash_success_msg,
                      inject_result)

        text = "----step4:网卡闪断过程中发布端更新数据 expect:成功----"
        self.log.info(text)
        sql_cmd = f"insert into {self.tb_name} values " \
            f"(generate_series(1,1000));"
        pub_msg = self.pub_sh.executDbSql(sql_cmd,
                                          cmd=self.user_param_pub)
        self.assertIn(self.constant.insert_success_msg, pub_msg, '执行失败'
                      + text)

        text = "----step5:订阅端查询数据是否更新 expect:数据更新----"
        self.log.info(text)
        time.sleep(40)
        sql_select = f"select count(*) from {self.tb_name};"
        self.log.info(sql_select)
        sub_msg = self.sub_sh.executDbSql(sql_select,
                                          dbname=None,
                                          cmd=self.user_param_sub,
                                          env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertEqual('1000', str(sub_msg.splitlines()[-2]).strip(),
                         '执行失败' + text)

        text = "----step6:订阅端注入网卡闪断故障 expect:成功----"
        self.log.info(text)
        self.log.info('----获取网卡----')
        network_cmd = f'ifconfig |grep {self.pri_sub_root.db_host} -B 1'
        self.log.info(network_cmd)
        network_msg = self.pri_sub_root.sh(network_cmd).result()
        network_card = network_msg.splitlines()[0].split(':')[0].strip()
        self.log.info(network_card)
        self.log.info("----故障注入----")
        inject_cmd = f"rNetLink_flash (repeat,duration,interval,dev) " \
            f"values(3,5,5,{network_card})"
        inject_result = self.com.cfe_inject(self.pri_sub_root, inject_cmd)
        self.log.info(inject_result)
        self.assertIn(self.constant.cfe_inject_netflash_success_msg,
                      inject_result)

        text = "----step7:网卡闪断过程中发布端更新数据 expect:成功----"
        self.log.info(text)
        sql_cmd = f"update {self.tb_name} set c_text='update';"
        pub_msg = self.pub_sh.executDbSql(sql_cmd,
                                          cmd=self.user_param_pub)
        self.assertIn(self.constant.update_success_msg, pub_msg, '执行失败'
                      + text)

        text = "----step8:订阅端查询数据是否更新 expect:数据更新----"
        self.log.info(text)
        time.sleep(280)
        sql_select1 = f"select c_text from {self.tb_name};"
        sub_msg = self.sub_sh.executDbSql(sql_select1,
                                          dbname=None,
                                          cmd=self.user_param_sub,
                                          env_path=macro.DB_ENV_PATH_REMOTE1)
        self.assertEqual(sub_msg.count('update'), 1000, '执行失败' + text)

    def tearDown(self):
        self.log.info("----this is teardown----")
        text = "----step9:清理环境 expect:成功----"
        self.log.info(text)
        self.log.info("--删除订阅--")
        sql_cmd = f"drop subscription if exists {self.sub_name};"
        drop_sub = self.sub_sh.executDbSql(sql_cmd,
                                          dbname=None,
                                          cmd=self.user_param_sub,
                                          env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info("--删除发布--")
        sql_cmd = f"drop publication if exists {self.pub_name};"
        drop_pub = self.pub_sh.executDbSql(sql_cmd,
                                           cmd=self.user_param_pub)
        self.log.info("--订阅端删除表--")
        sql_cmd = f"drop table if exists {self.tb_name};"
        drop_tb1 = self.sub_sh.executDbSql(sql_cmd,
                                          dbname=None,
                                          cmd=self.user_param_sub,
                                          env_path=macro.DB_ENV_PATH_REMOTE1)
        self.log.info("--发布端删除表--")
        drop_tb2 = self.pub_sh.executDbSql(sql_cmd,
                                           cmd=self.user_param_pub)
        self.log.info("--还原pg_hba配置文件--")
        cmd = f"mv {os.path.join(self.parent_path_pub, 'pg_hba.conf')} " \
            f"{os.path.join(macro.DB_INSTANCE_PATH, 'pg_hba.conf')} "
        self.log.info(cmd)
        result = self.pri_pub_dbuser.sh(cmd).result()
        self.log.info(result)
        self.log.info("--还原wal_level参数--")
        result = self.pub_sh.executeGsguc('set',
                                          self.constant.GSGUC_SUCCESS_MSG,
                                          f'wal_level={self.init_wal_level}')
        self.log.info(result)
        self.log.info("--重启数据库--")
        stop_msg = self.pub_sh.stopDbCluster()
        start_msg = self.pub_sh.startDbCluster()

        self.log.info(f"----{os.path.basename(__file__)} end----")
        self.assertTrue(stop_msg, '执行失败' + text)
        self.assertTrue(start_msg, '执行失败' + text)
        self.assertTrue(result, '执行失败' + text)
        self.assertIn(self.constant.drop_pub_succ_msg, drop_pub,
                      '执行失败' + text)
        self.assertIn(self.constant.drop_sub_succ_msg, drop_sub,
                      '执行失败' + text)
        self.assertIn(self.constant.DROP_TABLE_SUCCESS, drop_tb1, '执行失败'
                      + text)
        self.assertIn(self.constant.DROP_TABLE_SUCCESS, drop_tb2, '执行失败'
                      + text)
